package pageRank;

import java.io.IOException;

import mapred.AbstractMapper;
import mapred.util.Bytes;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

public class Readable_Mapper extends AbstractMapper<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {

	@Override
	public void map(BytesWritable key, BytesWritable value,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		double z = Double.parseDouble(this.getJobConf().get("Z"));
        
		ReadBytes v = new ReadBytes(value);
		double pagerank = v.nextDouble() + z;
		output.collect(key, new BytesWritable(Bytes.Double2Bytes(pagerank)));
	}

//	@Override
//	public void map(Text key, Text value, OutputCollector<Text, Text> output,
//			Reporter reporter) throws IOException {
//		double z = Double.parseDouble(this.getJobConf().get("Z"));
//		String str = value.toString();
//		int pos = str.indexOf("\t");
//		double pagerank = Double.parseDouble(str.substring(0, pos)) + z;
//		output.collect(key, new Text(String.valueOf(pagerank)));
//	}

}
